<?php
namespace Baihe\Kafkaservice;

//php版本对kafka兼容度不是很好问题，通过代理生产消息
//使用方式: KProxy::log($msg,$topic); 
//$msg => array or string | $topic => string
//写入kafka会自动添加一个当前时间字段，tik

class KProxyConf {
    public static $proxyServer = 'tcp://127.0.0.1';
    public static $proxyPort = '6029';
}

class KProxy {

    private $debug = false;
    private static $proxy = null;
    private $_socket = null;
    private $defaultPort = 6029;
    private $timeout = 1;

    private function __construct(){}
    private function __clone(){}

    public static function getProxy($group = 'baihe'){
        if(!isset(self::$proxy[$group])){
            self::$proxy[$group] = null;
        }
        if (!self::$proxy[$group] instanceof self) {
            self::$proxy[$group] = new self();
            self::$proxy[$group]->connect();
        }
        return self::$proxy[$group];
    }

    public function __destruct(){
        $this->disconnect();
    }

    /** 
     * 推送消息到kafka
     * @param $topic
     * @param $key
     * @param $msg
     * @return array|bool
     */
    public static function log($msg, $topic, $key='baihe'){
        if (empty($topic) || empty($key) || empty($msg)) {
            return false;
        }

        $time = date('Y-m-d H:i:s');
        if(is_array($msg)){
            $msg['tik'] = $time;
        }elseif(is_string($msg)){
            $msg .= ' · '.$time;
        }

        return self::sendMessage($topic, $msg, $key);
    }

    private static function sendMessage($topic, $msg, $key){
        $proxy = self::getProxy($key);
        if($proxy->connect()){
            $m = array("fun"=>'kafka', "s"=>'log', "t"=>$topic, "p"=>"-1", "k"=>$key, "v"=>$msg);
            $packet = json_encode($m,JSON_UNESCAPED_UNICODE);
            return $proxy->writeAll($packet);
        }
        return false;
    }

    public function writeAll($pack){
        $pack_len = strlen($pack);
        $pack = pack("sA{$pack_len}",$pack_len,$pack);
        $res = fwrite($this->_socket, $pack, strlen($pack));
        $this->disconnect();
        return empty($res) ? false : true ;
    }

    public function connect(){
        if($this->isConnected()){
            return true;
        }

        $errno = '';
        $errstr = '';
        $this->_socket = @pfsockopen(
                KProxyConf::$proxyServer, 
                KProxyConf::$proxyPort, 
                $errno, 
                $errstr, 
                $this->timeout
            );
        if(is_resource($this->_socket)){
            return true;
        }
        if(!empty($errstr)){
            echo $errstr."\n";
        }
        return false;
    }

    public function connected(){
        if(is_resource($this->_socket)){
            return true;
        }
        return false;
    }
    
    public function isConnected(){
        return is_resource($this->_socket);
    }

    public function disconnect(){
        if (is_resource($this->_socket)) {
            fclose($this->_socket);
        }
        $this->_socket = null;
    }

    public static function debug($val){
        if(!self::$debug){
            return false;
        }
        echo var_export($val,true)."\n";
    }
//end of class
}

//单独测试
if($argv[1] == "hihu"){

    for($i=0;$i<5;$i++){
        $rand = rand(1,10);
        $arr = [
            'addtime'=>date("Y-m-d H:i:s"),
            'url'=>'https://www.baihe.com/?s='.$rand,
            'uid'=>18728194,
            'name'=>'hihu',
            'platform'=>'1203',
            'reginfo'=>'info-'.$i,
            'fix'=>false,
            ];
        KProxy::log($arr,'hihutest'); //百合kafka，测试topic hihutest
        if($rand == 3){
           sleep(1);
        }
    }
}


//批量测试
if($argv[1] == "hihuall"){
    if(!\is_numeric($argv[2])){
        die("error argv2");
    }
    $start = intval($argv[2]);
    for($i=0;$i<100;$i++){
        $rand = rand(1,20);
        $arr = [
            'addtime'=>date("Y-m-d H:i:s"),
            'url'=>'https://www.baihe.com/?s='.$rand,
            'uid'=>18728194,
            'name'=>'feifei',
            'platform'=>'1203',
            'reginfo'=>'info-'.$start,
            'fix'=>false,
            ];
        KProxy::log($arr,'hihutest'); //百合kafka，测试topic hihutest
        if($rand == 3){
           sleep(1);
        }
        $start++;
    }
}
